package top.atcra.rpc.transport.common;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import top.atcra.rpc.RPCBootstrap;
import top.atcra.rpc.config.ConfigLoader;
import top.atcra.rpc.server.shutdown.ShutdownHook;
import top.atcra.rpc.service.ImplementRegistry;
import top.atcra.rpc.transport.message.RPCRequest;
import top.atcra.rpc.transport.message.RPCResponse;
import top.atcra.rpc.transport.message.RPCResponseCode;
import top.atcra.rpc.transport.message.RequestPayload;

import java.lang.reflect.InvocationTargetException;

/**
 * 该类型用于实现方法调用的逻辑。输入为Request，输出为Response
 * <p>
 * 主要实现以下功能点
 * 1. 服务关闭阶段的处理（基于挡板）
 * 2. 服务限流
 * 3. 心跳请求处理
 * 4. 常规请求处理
 */
@Slf4j
public class MethodCaller extends SimpleChannelInboundHandler<RPCRequest> {
    //    ExecutorService executorService = new ThreadPoolExecutor(10, 100, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000));
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RPCRequest request) throws Exception {
        Channel channel = ctx.channel();
        RPCResponse response = new RPCResponse().setRequestId(request.getRequestId())
                .setTimestamp(System.currentTimeMillis())
                .setCompressType(request.getCompressType())
                .setSerializeType(request.getSerializeType());
        var type = RPCRequest.RequestType.valueOf(request.getRequestType());
        // 服务关闭阶段的处理
        if (ShutdownHook.ENTER_DELAY_SHUTDOWN_STATE.get()) {
            response.setCode(RPCResponseCode.CLOSING.getCode());
        } else {
            ShutdownHook.REQUEST_COUNTER.increment();
            if (type == RPCRequest.RequestType.HEALTH_DETECT) {
                response.setCode(RPCResponseCode.HEARTBEAT_SUCCESS.getCode());
            } else {
                // 服务限流
                var rateLimiter = RPCBootstrap.RATE_LIMITERS.get(channel.remoteAddress());
                if (rateLimiter == null) {
                    rateLimiter = ConfigLoader.getConfig().getRateLimiter();
                    RPCBootstrap.RATE_LIMITERS.put(channel.remoteAddress(), rateLimiter);
                }
                if (!rateLimiter.allowEnter()) {
                    response.setCode(RPCResponseCode.LIMIT.getCode());
                    log.warn("编号为【{}】的请求被限流", request.getRequestId());
                } else {
                    try {
                        Object result = invoke(request.getBody());
                        if (log.isDebugEnabled()) {
                            log.debug("请求【{}】已经在服务端完成方法调用。", request.getRequestId());
                        }
                        response.setCode(RPCResponseCode.SUCCESS.getCode());
                        response.setBody(result);
                    } catch (NoSuchMethodException e) {
                        log.error("编号为【{}】的请求在调用过程中发生异常。目标方法{}不存在", request.getRequestId(), request.getBody().getMethodName(), e);
                        response.setCode(RPCResponseCode.METHOD_NOT_FOUND.getCode());
                        response.setBody(request.getBody().getMethodName());
                    } catch (Exception e) {
                        log.error("编号为【{}】的请求在调用过程中发生异常。", request.getRequestId(), e);
                        response.setCode(RPCResponseCode.FAIL.getCode());
                        response.setBody(e.getMessage());
                    }
                }
            }
            ShutdownHook.REQUEST_COUNTER.decrement();
        }

        channel.writeAndFlush(response);
    }

    /**
     * 服务端执行方法调用
     *
     * @param payload 请求负载
     * @return 响应负载
     */
    private Object invoke(RequestPayload payload) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException, InstantiationException {
        var instance = ImplementRegistry.get(payload.getServiceName() + payload.getGroup());
        var method = instance.getClass().getMethod(payload.getMethodName(), payload.getParametersType());
        return method.invoke(instance, payload.getParametersValue());
    }
}
